package com.atguigu.gmall.realtime.app.newapp

import com.atguigu.gmall.realtime.app.newapp.DwOrderWideNewApp.sparkConf
import com.atguigu.gmall.realtime.bean.{FavorInfo, OrderRefundInfo}
import com.atguigu.gmall.realtime.handler.{CommonApp, DimHandler, EsHandler}
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.DStream

import java.time.LocalDate

object DwRefundApp extends CommonApp {

  def main(args: Array[String]): Unit = {
    val groupId: String = "dw_refund_info_app"
    val orderRefundTopic: String = "DWD_ORDER_REFUND_INFO_I"

    val ssc: StreamingContext = initSsc(groupId, "local[4]", 5)

    val orderRefundStream: DStream[OrderRefundInfo] = getDstream[OrderRefundInfo](ssc, orderRefundTopic, groupId)


    val dataWithUDstream: DStream[OrderRefundInfo] = DimHandler.joinUser[OrderRefundInfo](orderRefundStream,
      { orderRefundInfo: OrderRefundInfo => orderRefundInfo.user_id.toString }
    )

    val dataWithSkuDStream: DStream[OrderRefundInfo] = DimHandler.joinSku[OrderRefundInfo](dataWithUDstream,
      { orderRefundInfo: OrderRefundInfo => orderRefundInfo.sku_id.toString }
    )

    val dataWithRefundTypeDStream: DStream[OrderRefundInfo] = DimHandler.joinBaseDic[OrderRefundInfo](dataWithSkuDStream,
      { orderRefundInfo: OrderRefundInfo => orderRefundInfo.refund_type },
      { (orderRefundInfo, dicName) => orderRefundInfo.refund_type_name = dicName }
    )

    val dataWithReasonTypeDStream: DStream[OrderRefundInfo] = DimHandler.joinBaseDic[OrderRefundInfo](dataWithRefundTypeDStream,
      { orderRefundInfo: OrderRefundInfo => orderRefundInfo.refund_reason_type },
      { (orderRefundInfo, dicName) => orderRefundInfo.refund_reason_type_name = dicName }
    )

    val refundDtDstream: DStream[OrderRefundInfo] = dataWithReasonTypeDStream.map { objInfo =>
      val dateTimeArr: Array[String] = objInfo.create_time.split(" ")
      objInfo.create_date = dateTimeArr(0)
      val timeArr: Array[String] = dateTimeArr(1).split(":")
      objInfo.create_hour = timeArr(0)
      objInfo
    }

    refundDtDstream.cache()
    refundDtDstream.print(1000)



    EsHandler.saveDStreamToEs[OrderRefundInfo](refundDtDstream,
      "gmall2022_refund_info_",
      "id",
      {Unit=>this.commitAllOffsetKafka()}
    )


    ssc.start()
    ssc.awaitTermination()

  }


}
